package com.yc.geo.geohash.udf;

import com.google.common.collect.Lists;
import com.hive.udf.util.JsonUtils;
import com.hive.udf.util.VincentyGeodesy;
import com.yc.geo.geohash.model.Region;
import com.yc.geo.geohash.model.WGS84Point;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;
import org.codehaus.jackson.map.annotate.JsonSerialize;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.List;
/**
 * 转换经纬度为 "能和墨迹天气对应上的区域" : eg: 39.465226,121.562056 => 辽宁省普兰店区
 * 注： 按距离每个区的距离算，离哪个区中心最近返回哪个区
 * 如果需要准确返回"归属行政区域"，需要调用高德API，效率较低
 * @author zhaoxiang
 */
public class ReGeoForMoJi extends UDF{
	private static Logger logger = LoggerFactory.getLogger(ReGeoForMoJi.class);
	private static ObjectMapper objectMapper = new ObjectMapper();
	private static final List<Region> REGIONS = Lists.newLinkedList();
	private static final List<String> REGIONS_MOJI = Lists.newLinkedList();
	static {
		objectMapper.setSerializationInclusion(JsonSerialize.Inclusion.NON_NULL); 
		objectMapper.configure(SerializationConfig.Feature.WRITE_DATES_AS_TIMESTAMPS, false);
		objectMapper.getDeserializationConfig().withDateFormat(new SimpleDateFormat("yyyy-MM-dd"));
		objectMapper.getSerializationConfig().withDateFormat(new SimpleDateFormat("yyyy-MM-dd"));
		objectMapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
		objectMapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
		objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true);
		objectMapper.configure(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS, false);
		init();
	}
	
	public static void init() {
		LineIterator lineIterator;
		try {
			//加载行政区域数据
			lineIterator = IOUtils.lineIterator(ReGeoForMoJi.class.getClassLoader().getResourceAsStream("region-cn.txt"), "UTF-8");
			while(lineIterator.hasNext()){
				String l = lineIterator.nextLine();
				Region region = JsonUtils.fromJSON(l, Region.class);
				REGIONS.add(region);
			}
			lineIterator.close();
			
			//加载行政区域数据
			lineIterator = IOUtils.lineIterator(ReGeoForMoJi.class.getClassLoader().getResourceAsStream("moji-district.txt"), "UTF-8");
			while(lineIterator.hasNext()){
				REGIONS_MOJI.add(lineIterator.nextLine());
			}
			lineIterator.close();
			
		} catch (IOException e) {
			logger.error("load region-cn.txt error:", e);
			System.exit(127);
		}	
	}
	
	/**
	 * @param latitude 纬度
	 * @param longitude 经度
	 * @return 省，市，区，  eg： 广东省,梅州市,五华县
	 */
	public String evaluate(final Double latitude, final Double longitude) {  
        try {
        	if(latitude == null || longitude == null) {
        		return "-";
        	}
        	Region curRegion = null;
			WGS84Point curPoint = new WGS84Point(latitude, longitude);
			double minDistance = Double.MAX_VALUE;
			for(Region region : REGIONS){
				double lat1 = region.getLat();
				double lon1 = region.getLon();
				//经纬度都相差不多，才计算距离
				if(Math.abs(lat1 - latitude) < 0.4 && Math.abs(lon1 - longitude) < 0.5){
					WGS84Point otherPoint = new WGS84Point(lat1, lon1);
					double distance = VincentyGeodesy.distanceInMeters(curPoint, otherPoint);
					if(distance < minDistance){
						minDistance = distance;
						curRegion = region;
					}
				}
			}
			if (curRegion == null) {
				return "-";
			}
			
			String province = curRegion.getProvince();
			String city = curRegion.getCity();
			String district = curRegion.getDistrict();
			
			String address = province + district;
			if(! REGIONS_MOJI.contains(address)){
				address = province + city + district;
				if(! REGIONS_MOJI.contains(address)){
					address = province + city;
				}
			}
        	return address;
		} catch (Exception e) {
			logger.error("get district error, lat: {}, lon:{}", latitude, longitude);
			logger.error("",e);
			return "-";
		}
    }  
	
	public static void main(String[] args) throws IOException {
		ReGeoForMoJi reGeo = new ReGeoForMoJi();
		System.out.println(reGeo.evaluate(22.303095 ,114.16728));
		long start = System.currentTimeMillis();
//		//加载天气数据
//		Map<String,String> weatherMap = Maps.newHashMap();
//		LineIterator lineIterator = IOUtils.lineIterator(ReGeo.class.getClassLoader().getResourceAsStream("weather.txt"), "UTF-8");	
//		while(lineIterator.hasNext()){
//			String jsonStr = lineIterator.nextLine();
//			JsonNode jsonNode =  objectMapper.readTree(jsonStr);
//			String province_district = jsonNode.get("city").asText() + jsonNode.get("area").asText();
//			weatherMap.put(province_district,jsonStr);
//		}
//		lineIterator.close();
//		
//		PrintWriter writer = new PrintWriter("src/main/resources/result.txt");
//		//读订单数据
//		lineIterator = IOUtils.lineIterator(ReGeo.class.getClassLoader().getResourceAsStream("order.txt"), "UTF-8");
//		int i=0;
//		while(lineIterator.hasNext()){
//			String orderLine = lineIterator.nextLine();
//			String[] fields = orderLine.split("\t");
//			String service_order_id = fields[0];
//			String city_en = fields[18];
//			double start_latitude = Double.parseDouble(fields[49]);
//			double start_longitude = Double.parseDouble(fields[50]);
//			Region curRegion = null;
//			
//			WGS84Point curPoint = new WGS84Point(start_latitude, start_longitude);
//			double minDistance = Double.MAX_VALUE;
//			for(Region region : REGIONS){
//				double lat1 = region.getLat();
//				double lon1 = region.getLon();
//				//经纬度都相差不多，才计算距离
//				if(Math.abs(lat1 - start_latitude) < 0.4 && Math.abs(lon1 - start_longitude) < 0.4){
//					WGS84Point otherPoint = new WGS84Point(lat1, lon1);
//					double distance = VincentyGeodesy.distanceInMeters(curPoint, otherPoint);
//					if(distance < minDistance){
//						minDistance = distance;
//						curRegion = region;
//					}
//				}
//			}
//			if (curRegion == null) {
//				System.err.println(String.format("%s,%s,%s,%s",GeoHash_V2.encode(start_latitude, start_longitude, 7).toHashString(),city_en, start_latitude,start_longitude));
//				i++;
//				continue;
//			}
//			
//			String province = curRegion.getProvince();
//			String city = curRegion.getCity();
//			String district = curRegion.getDistrict();
//			String weather = weatherMap.get(province + district);
//			if(weather == null){
//				weather = weatherMap.get(province + city + district);
//				if(weather == null){
//					weather = weatherMap.get(province + city);
//				}
//			}
//			String[] rs =new String[]{service_order_id, city_en,start_latitude +"",start_longitude+"",province,city, district,weather};
//			writer.println(StringUtils.join(rs, ","));
//		}
//		
//		writer.close();
//		System.out.println("totail not fount " + i);
		
		long end = System.currentTimeMillis();
		System.out.println("totail time " + (end - start) +"ms !");
		
	}
}
